package com.zhenjun.dimension;

import com.alibaba.fastjson.JSONObject;
import com.zhenjun.domin.Order;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.calcite.shaded.com.google.common.cache.Cache;
import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author wangzj
 * @description 预加载全量数据, 每5分钟获取一次
 * @date 2020/8/17 23:45
 */
public class WholeLoad extends RichMapFunction<String, Order> {
    private static final Logger log = LoggerFactory.getLogger(WholeLoad.class);
    //定时任务执行器
    ScheduledExecutorService executor = null;
//    private Map<String, String> cache = null;
     Cache<String, String> cache = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //每5分钟启动一个线程加载数据库中的数据
        executor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    load();
                } catch (Exception e) {
                    e.printStackTrace();
                    e.getMessage();
                }
            }
        }, 5, 5, TimeUnit.MINUTES);
        //1、执行线程  2、初始化延时5m  3、两次开始执行最小间隔时间5m 4、计时单位分钟
    }

    @Override
    public Order map(String value) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        Integer cityId = jsonObject.getInteger("city_id");
        String userName = jsonObject.getString("user_name");
        String items = jsonObject.getString("items");
//        String cityName = cache.get(cityId);
        String cityName = cache.getIfPresent(cityId);

        return new Order(cityId, cityName, userName, items);
    }

    /**
     * 从数据库中查询数据
     *
     * @throws Exception
     */
    public void load() throws Exception {
        log.warn("+++++++连接数据库，获取城市名称+++++");
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://47.104.87.133:3306/dim?characterEncoding=UTF-8", "wangzj", "123456");
        String sql = "select city_id,city_name from city_info";
        PreparedStatement ps = con.prepareStatement(sql);
        ResultSet rs = ps.executeQuery();
        while (rs.next()) {
            String cityId = rs.getString("city_id");
            String cityName = rs.getString("city_name");
            cache.put(cityId, cityName);
        }
//        rs.close();
//        ps.close();
        con.close();
    }
}
